package SparkChat

import java.util.UUID

import SparkChat.CommonInfo.{HearBeat, ResignInfo, ResignResultMsg, startMsg}
import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

import scala.concurrent.duration.DurationInt

class SparkWorker(host:String,port:Int) extends Actor{
  var masterProxy: ActorSelection = _


  val id = UUID.randomUUID().toString
  override def receive: Receive = {
    case startMsg(msg)=>{
      println("发起注册")
      masterProxy ! ResignInfo(id)
    }
    case ResignResultMsg(id)=>{
      import context.dispatcher
      println("收到返回的注册成功信息")
      context.system.scheduler.schedule(0 millis,3000 millis,self,HearBeat(id))
    }
    case HearBeat(id)=>
      println(s"$id work 发起心跳检测")
      masterProxy ! HearBeat(id)

  }

  override def preStart(): Unit = {
    super.preStart()
    masterProxy = context.actorSelection(s"akka.tcp://SparkMaster@${host}:${port}/user/SparkMaster")

  }
}

object SparkWorker extends App{
  val (masterHost, masterPort, workerHost, workerPort) = ("127.0.0.1", 10002, "127.0.0.1", 10001)
  val config = ConfigFactory.parseString(
    s"""
       |akka.actor.provider="akka.remote.RemoteActorRefProvider"
       |akka.remote.netty.tcp.hostname=$workerHost
       |akka.remote.netty.tcp.port=$workerPort
      """.stripMargin
  )

  val sparkWorkerSystem = ActorSystem("SparkWorker", config)
  val sparkActor= sparkWorkerSystem.actorOf(Props(new SparkWorker(masterHost, masterPort)), "SparkWorker")

  sparkActor ! startMsg(s"host:$workerHost port:$workerPort")
}
